package org.pizza.rocket.core.producer;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.SendCallback;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.bean.OrderProducerBean;
import com.aliyun.openservices.ons.api.bean.ProducerBean;
import com.aliyun.openservices.ons.api.bean.TransactionProducerBean;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionChecker;
import com.aliyun.openservices.ons.api.transaction.LocalTransactionExecuter;
import com.aliyun.openservices.ons.api.transaction.TransactionStatus;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.pizza.rocket.core.properties.RocketMqProperties;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.function.Consumer;

/**
 * @author 高巍
 * email kynb_club@163.com
 * @since 2020/11/24 9:51 上午
 */
public class RocketMQProducer implements InitializingBean, DisposableBean, ApplicationContextAware {
    private static final Logger log = LoggerFactory.getLogger(RocketMQProducer.class);
    private ApplicationContext applicationContext;
    private ObjectMapper objectMapper;
    private ProducerBean defaultProducer;

    private OrderProducerBean orderProducer;
    private RocketMqProperties rocketProperties;
    /**
     * only put TransactionProducerBean by now!!!
     */
    private final Map<String, TransactionProducerBean> cache = new ConcurrentHashMap<>();

    @Override
    public void destroy() throws Exception {
        if (Objects.nonNull(defaultProducer)) {
            defaultProducer.shutdown();
        }
        if (Objects.nonNull(orderProducer)) {
            orderProducer.shutdown();
        }
        for (Map.Entry<String, TransactionProducerBean> kv : cache.entrySet()) {
            if (Objects.nonNull(kv.getValue())) {
                kv.getValue().shutdown();
            }
        }
        cache.clear();
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        if (defaultProducer != null) {
            defaultProducer.start();
        }
        if (orderProducer != null) {
            orderProducer.start();
        }
    }


    public ObjectMapper getObjectMapper() {
        return objectMapper;
    }

    public void setObjectMapper(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    public ProducerBean getDefaultProducer() {
        return defaultProducer;
    }

    public void setDefaultProducer(ProducerBean defaultProducer) {
        this.defaultProducer = defaultProducer;
    }

    public OrderProducerBean getOrderProducer() {
        return orderProducer;
    }

    public void setOrderProducer(OrderProducerBean orderProducer) {
        this.orderProducer = orderProducer;
    }

    public RocketMqProperties getRocketProperties() {
        return rocketProperties;
    }

    public void setRocketProperties(RocketMqProperties rocketProperties) {
        this.rocketProperties = rocketProperties;
    }

    /**
     * 同步发送消息，只要不抛异常就表示成功
     *
     * @param message 要发送的消息对象
     * @return 发送结果，含消息Id, 消息主题
     */
    public SendResult send(final Message message) {
        return defaultProducer.send(message);
    }

    public SendResult sendOrderly(final Message message,final String shardingKey ){
        return orderProducer.send(message, shardingKey);
    }
    /**
     * 发送消息，Oneway形式，服务器不应答，无法保证消息是否成功到达服务器
     *
     * @param message 要发送的消息
     */
   public void sendOneway(final Message message) {
        defaultProducer.sendOneway(message);
    }

    /**
     * 发送消息，异步Callback形式
     *
     * @param message      要发送的消息
     * @param sendCallback 发送完成要执行的回调函数
     */
    public void sendAsync(final Message message, final SendCallback sendCallback) {
        defaultProducer.sendAsync(message, sendCallback);
    }

    /**
     * 设置异步发送消息执行Callback的目标线程池。
     * <p>
     * 如果不设置，将使用公共线程池，仅建议执行轻量级的Callback任务，避免阻塞公共线程池
     * 引起其它链路超时。
     *
     * @param callbackExecutor 执行Callback的线程池
     */
    public void setCallbackExecutor(final ExecutorService callbackExecutor) {
        defaultProducer.setCallbackExecutor(callbackExecutor);
    }

    /**
     * 该方法用来发送一条事务型消息. 一条事务型消息发送分为三个步骤:
     * <ol>
     *     <li>本服务实现类首先发送一条半消息到到消息服务器;</li>
     *     <li>通过<code>executer</code>执行本地事务执行器;</li>
     *     <li>根据上一步骤执行结果, 决定发送提交或者回滚第一步发送的半消息;</li>
     * </ol>
     *
     * @param message 要发送的事务型消息
     * @param executer 执行本地事务执行器
     * @param arg     应用自定义参数，该参数可以传入本地事务执行器
     * @return 发送结果.
     */
    public SendResult sendInTransaction(String producerGroup, final Message message, final LocalTransactionExecuter executer, final Object arg) {
        if (!cache.containsKey(producerGroup)) {
            throw new RuntimeException("[" + producerGroup + "]没有找到TransactionProducerBean");
        }
        TransactionProducerBean transactionProducerBean = cache.get(producerGroup);
        return transactionProducerBean.send(message, executer, arg);
    }

    /**
     * 该方法用来发送一条事务型消息. 一条事务型消息发送分为三个步骤:
     * <ol>
     *     <li>本服务实现类首先发送一条半消息到到消息服务器;</li>
     *     <li>通过<code>executer</code>执行本地事务执行器;</li>
     *     <li>根据上一步骤执行结果, 决定发送提交或者回滚第一步发送的半消息;</li>
     * </ol>
     *
     * @param message  要发送的事务型消息
     * @param consumer 应用自定义本地执行逻辑
     * @return 发送结果.
     */
    public SendResult sendInTransaction(String producerGroup, final Message message, Consumer<Message> consumer) {
        if (!cache.containsKey(producerGroup)) {
            throw new RuntimeException("[" + producerGroup + "]没有找到TransactionProducerBean");
        }
        TransactionProducerBean transactionProducerBean = cache.get(producerGroup);
        return transactionProducerBean.send(message, (msg, arg) -> {
            try {
                consumer.accept(msg);
            }catch (Exception e){
                log.error("事务执行异常 :{}", e.getMessage());
                return TransactionStatus.RollbackTransaction;
            }
            return TransactionStatus.CommitTransaction;
        }, null);
    }


    /**
     * 创建并启动一个事务消息生产者
     * @param producerGroup
     * @param checker
     * @param properties
     * @return
     */
    public TransactionProducerBean createAndStartTransactionProducer(String producerGroup, LocalTransactionChecker checker, Properties properties) {
        producerGroup = applicationContext.getEnvironment().resolveRequiredPlaceholders(producerGroup);
        if (cache.containsKey(producerGroup)) {
            log.info(String.format("get TransactionProducerBean '%s' from cache", producerGroup));
            return null;
        }
        TransactionProducerBean transactionProducerBean = new TransactionProducerBean();
        transactionProducerBean.setProperties(properties);
        transactionProducerBean.setLocalTransactionChecker(checker);
        //start
        transactionProducerBean.start();
        cache.put(producerGroup, transactionProducerBean);
        return transactionProducerBean;
    }

    public void removeTransactionProducer(String producerGroup) {
        if (cache.containsKey(producerGroup)) {
            TransactionProducerBean transactionProducerBean = cache.get(producerGroup);
            transactionProducerBean.shutdown();
            cache.remove(producerGroup);
        }
    }
}
